package io.gitee.kenewstar.dbmq;

import cn.hutool.core.collection.CollUtil;
import io.gitee.kenewstar.dbmq.data.DbDataHandler;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author kenewstar
 */
public class ConsumerTask {

    private List<Long> downIds;

    private Long oldId;

    private Long id;

    private String topic;

    private Integer tag;

    private Integer limit;

    private String uuid;

    private ConsumeStatus defaultStatus;

    private DbMqConsumerInf consumerInf;

    public static ConsumerTask build(DbMqConsumerInf consumerInf, Long id, String uuid) {
        ConsumerTask consumerTask = new ConsumerTask();
        final DbMqConsumer sqlConsumer = consumerInf.getClass().getAnnotation(DbMqConsumer.class);
        consumerTask.topic = sqlConsumer.topic();
        consumerTask.limit = sqlConsumer.batchCount();
        consumerTask.defaultStatus = sqlConsumer.defaultConsumeStatus();
        consumerTask.consumerInf = consumerInf;
        consumerTask.id = id;
        consumerTask.uuid = uuid;
        consumerTask.downIds = new ArrayList<>();
        return consumerTask;
    }

    public void executeTask(DbDataHandler dbDataHandler) {
        final List<Message> messageList = dbDataHandler.getMessage(this);
        if (CollUtil.isNotEmpty(messageList)) {
            messageList.forEach(x -> x.setConsumeFlag(defaultStatus.value()));
            consumerInf.consume(messageList);
        }

        final List<Long> msgIds = messageList.stream()
                .filter(message -> message.getConsumeFlag() == 1)
                .map(Message::getId).collect(Collectors.toList());
        dbDataHandler.deleteMessage(topic, msgIds);

    }

    public List<Long> getDownIds() {
        return downIds;
    }

    public void setDownIds(List<Long> downIds) {
        this.downIds = downIds;
    }

    public Long getOldId() {
        return oldId;
    }

    public void setOldId(Long oldId) {
        this.oldId = oldId;
    }

    public Long getId() {
        return id;
    }

    public void setId(Long id) {
        this.id = id;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public Integer getTag() {
        return tag;
    }

    public void setTag(Integer tag) {
        this.tag = tag;
    }

    public Integer getLimit() {
        return limit;
    }

    public void setLimit(Integer limit) {
        this.limit = limit;
    }

    public String getUuid() {
        return uuid;
    }

    public void setUuid(String uuid) {
        this.uuid = uuid;
    }
}
